package com.link.base.base.consumer.service;

import com.alibaba.fastjson.JSONObject;
import com.link.base.base.accnt.dao.mybatis.mapper.AcctAddressMapper;
import com.link.base.base.accnt.model.Account;
import com.link.base.base.accnt.model.AcctAddress;
import com.link.base.base.consumer.dao.mybatis.mapper.PortalAccntLineMapper;
import com.link.base.base.consumer.model.PortalAccntLine;
import com.link.base.base.consumer.model.PortalAccntTag;
import com.link.base.base.user.service.UserServiceImpl;
import com.link.base.core.basic.interceptors.mybatis.system.KafkaSender;
import com.link.base.user.model.CoreUser;
import com.link.core.basic.dao.KeyGenerateDao;
import com.link.core.basic.dao.mapper.BasicMapper;
import com.link.core.basic.service.BasicServiceImpl;
import com.link.core.basic.service.ServiceException;
import com.link.core.util.DateUtil;
import com.link.core.util.JsonUtil;
import com.link.core.util.StringUtils;
import com.link.core.util.UserUtil;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Date;

/**
 * 消费者列表
 * @author  yipeng.zhu@hand-china.com
 * @date:   2018/11/5 16:24
 * @version 1.0
 */

@Service
public class PortalAccntLineServiceImpl extends BasicServiceImpl<PortalAccntLine> implements PortalAccntLineService {
    @Resource
    private PortalAccntLineMapper portalAccntLineMapper;
    @Resource
    private AcctAddressMapper acctAddressMapper;
    @Resource
    private KeyGenerateDao keyGenerateDao;
    @Resource
    private PortalAccntTagService portalAccntTagService;
    @Resource
    private UserServiceImpl userService;

    @Override
    public BasicMapper<PortalAccntLine> getBasicMapper() throws Exception {
        return portalAccntLineMapper;

    }
    /**
     * 是存在消费者
     * @param       record
     * @return
     */

    @Override
    public PortalAccntLine queryAccntByCombinCondition(PortalAccntLine record) {
        return portalAccntLineMapper.queryAccntByCombinCondition(record);
    }

    @Override
    @Transactional(rollbackFor = ServiceException.class)
    public void moreModelInsert(PortalAccntLine accntline) throws Exception {
        CoreUser landedUser = UserUtil.getUser();
        //插入消费者
        if (StringUtils.isBlank(accntline.getFirstEngageTime())) {
            String engageTime = DateUtil.dateToStr(new Date(), "yyyy-MM-dd HH:mm:ss");
            accntline.setFirstEngageTime(engageTime);
        }
        portalAccntLineMapper.insert(accntline);
        //插入消费者地址
        if (accntline.getAccntTagList() != null && !accntline.getAddrList().isEmpty()) {
            for (AcctAddress model : accntline.getAddrList()) {
                model.setAcctId(accntline.getId());
                model.setId(keyGenerateDao.keyGenerate());
                acctAddressMapper.insert(model);
            }
        }
        //插入消费者标签
        if (accntline.getAccntTagList() != null && !accntline.getAccntTagList().isEmpty()) {
            for (PortalAccntTag model : accntline.getAccntTagList()) {
                model.setAccntRowId(accntline.getId());
                model.setOrgId(landedUser.getOrgId());
            }
            portalAccntTagService.batchUpsert(accntline.getAccntTagList());
        }
        Account account = JSONObject.toJavaObject(JSONObject.parseObject(JsonUtil.toString(accntline)), Account.class);
        //account.setDataSource2("Link");
        Object object = userService.transformToDmpConsumer(account);
        KafkaSender.sendInfo("CDC_CONSUMER", JsonUtil.toString(object));
    }


}
